package com.example.mq.mqserver.datacenter;

import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.Binding;
import com.example.mq.mqserver.core.Exchange;
import com.example.mq.mqserver.core.MSGQueue;
import com.example.mq.mqserver.core.Message;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * 使用这个类来统一管理内存中的所有数据
 */
public class MemoryDataCenter {
    //key是exchangeName, value是exchange对象
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    //key是queueName value是MSGQueue对象
    private ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();
    //key是exchangeName 第二个key是queueName
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();
    //key是messageId value是一个Message对象
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
    //key是queueName value是一个Message的链表
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> queueMessageWaitAckMap = new ConcurrentHashMap<>();

    public void insertExchange(Exchange exchange) {
        exchangeMap.put(exchange.getName(), exchange);
        System.out.println("[MemoryDataCenter] 新交换机添加成功 exchangeName = " + exchange.getName());

    }

    public Exchange getExchange(String exchangeName) {
        return exchangeMap.get(exchangeName);
    }

    public void deleteExchange(String exchangeName) {
        exchangeMap.remove(exchangeName);
        System.out.println("[MemoryDataCenter] 交换机删除成功 exchangeName = " + exchangeName);
    }

    public void insertQueue(MSGQueue queue) {
        queueMap.put(queue.getName(), queue);
        System.out.println("[MemoryDataCenter] 新队列添加成功 queueName = " + queue.getName());
    }

    public MSGQueue getQueue(String queueName) {
        return queueMap.get(queueName);
    }

    public void deleteQueue(String queueName) {
        queueMap.remove(queueName);
        System.out.println("[MemoryDataCenter]  队列删除成功 queueName = " + queueName);
    }

    public void insertBinding(Binding binding) throws MqException {
//        //先使用exchangeName来查找哈希表是否有存在
//        ConcurrentHashMap<String, Binding> bindingMap = this.bindingsMap.get(bingding.getExchangeName());
//        if (bindingMap == null){
//            bindingsMap.put(binding.getExchangeName(),bindingMap);
//        }
        //现根据exchangeName查一下 对应的哈希表是否存在 不存在就创建一个
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                k -> new ConcurrentHashMap<>());

        synchronized (bindingMap) {
            //再根据queueName查找一下 如果已经存在 就抛出异常 不存在才能插入
            if (bindingsMap.get(binding.getQueueName()) != null) {
                throw new MqException("[MemoryDataCenter] 绑定已经存在 exchangeName = " + binding.getExchangeName() +
                        ", queueName = " + binding.getQueueName());
            }
            bindingMap.put(binding.getQueueName(), binding);
        }
        System.out.println("[MemoryDataCenter] 新绑定添加成功 exchangeName = " + binding.getExchangeName() +
                ",queueName" + binding.getQueueName());
    }

    //获取绑定
    //1. 根据exchangeName和queueName确定唯一一个Binding
    //2. 根据exchangeName获取到所有的binding
    public Binding getBinding(String exchangeName, String queueName) {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if (bindingMap == null) {
            return null;
        }
        return bindingMap.get(queueName);
    }

    public ConcurrentHashMap<String, Binding> getBindings(String exchangeName) {
        return bindingsMap.get(exchangeName);
    }

    //删除绑定
    public void deleteBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if (bindingMap == null) {
            //该交换机没有绑定任何队列
            throw new MqException("[MemoryDataCenter] 绑定不存在 exchangeName = " + binding.getExchangeName() +
                    ", queueName = " + binding.getQueueName());
        }
        bindingMap.remove(binding.getQueueName());
        System.out.println("[MemoryDataCenter] 删除绑定成功 exchangeName = " + binding.getExchangeName() +
                ",queueName" + binding.getQueueName());
    }

    //添加消息
    public void addMessage(Message message) {
        messageMap.put(message.getMessageId(), message);
        System.out.println("[MemoryDataCenter] 新消息添加成功 messageId = " + message.getMessageId());
    }

    //根据id查消息
    public Message getMessage(String messageId) {
        return messageMap.get(messageId);
    }

    //根据id删消息
    public void removeMessage(String messageId) {
        messageMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息被移除 messageId = " + messageId);
    }

    //发送消息到指定队列
    public void sendMessage(MSGQueue queue, Message message) {
        //先把消息放在指定数据结构中
        //现根据队列的名称 找到该队列对应的消息链表
//        LinkedList<Message> messages = queueMessageMap.get(queue.getName());
//        if (messages == null) {
//            messages = new LinkedList<>();
//            queueMessageMap.put(queue.getName(), messages);
//        }
        LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queue.getName(), k -> new LinkedList<>());
        //把新消息添加到message中
        synchronized (messages){
            messages.add(message);
        }
        //在这里把该消息向消息中心插入
        addMessage(message);
        System.out.println("[MemoryDataCenter] 消息被投递到队列中 messageId = " + message.getMessageId());
    }
    //从队列中取消息
    public Message pollMessage(String queueName){
        //根据队列名 查找对应的消息链表
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        //如果没找到 则说明队列中没有任何消息
        if(messages == null){
            return null;
        }
        synchronized (messages){
            if(messages.size() == 0){
                return null;
            }
            //链表中有元素 就进行头删 取出该元素
            Message currentMessage = messages.remove(0);
            System.out.println("[MemoryDataCenter] 消息从队列中取出 messageId = " + currentMessage.getMessageId());
            return currentMessage;
        }
    }
    //获取指定队列中消息的个数

    public int getMessageCount(String queueName){
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if(messages == null){
            //队列中无消息
            return 0;
        }
        synchronized (messages){
            return messages.size();
        }
    }

    //未确认消息的操作
    //添加未确认的消息
    public void addMessageWaitAck(String queueName, Message message){
        ConcurrentMap<String,Message> messageMap = queueMessageWaitAckMap.computeIfAbsent(queueName,
                k -> new ConcurrentHashMap<>());
        messageMap.put(message.getMessageId(),message);
        System.out.println("[MemoryDataCenter] 消息进入到待确认队列 messageId=" + message.getMessageId());
    }
    //删除未确认的消息
    public void removeMessageWaitAck(String queueName, String messageId){
        ConcurrentHashMap<String, Message> messageMap = queueMessageWaitAckMap.get(queueName);
        if(messageMap == null){
            return;
        }
        messageMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息从待确认队列删除 messageId=" + messageId);
    }

    //获取指定未确认的消息
    public Message getMessageWaitAck(String queueName, String messageId){
        ConcurrentHashMap<String, Message> messageMap = queueMessageWaitAckMap.get(queueName);
        if(messageMap == null){
            return null;
        }
        return messageMap.get(messageId);
    }

    //当服务器重启后 要从硬盘上读取数据 回复到内存中
    public void recovery(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException {
        //清空所有旧数据
        exchangeMap.clear();
        queueMap.clear();
        bindingsMap.clear();
        messageMap.clear();
        queueMessageMap.clear();
        //恢复所有交换机
        List<Exchange> exchanges = diskDataCenter.selectAllExchanges();
        for (Exchange exchange : exchanges){
            exchangeMap.put(exchange.getName(),exchange);
        }
        //恢复所有队列
        List<MSGQueue> queueList = diskDataCenter.selectAllQueues();
        for(MSGQueue queue : queueList){
            queueMap.put(queue.getName(),queue);
        }
        //恢复所有绑定
        List<Binding> bindingList = diskDataCenter.selectAllBindings();
        for(Binding binding : bindingList){
            ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(), k -> new ConcurrentHashMap<>());
            bindingMap.put(binding.getQueueName(),binding);
        }
        //恢复所有消息
        //遍历所有的队列 根据每个队列的名字 获取到所有的信息
        for(MSGQueue queue : queueList){
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue(queue.getName());
            //将消息加载到内存
            queueMessageMap.put(queue.getName(),messages);
            for(Message message : messages){
                messageMap.put(message.getMessageId(),message);
            }
        }
        //未确认的消息 不需要从硬盘获取
        //在等待ack的过程中 未被确认的消息就转变为未被取走的消息
    }
}

